# coding=utf-8
"""
Author  : Jane
Contact : xijian@ict.ac.cn
Time    : 2021/3/26 13:27
Desc:
"""
# python 连接kafka 发送消息
import json
import sys
import time
import random
from kafka  import KafkaProducer
from kafka.errors import KafkaTimeoutError

class Producer(object):
    def __init__(self, KafkaServerList=['127.0.0.1:9092'], ClientId="Producer01", Topic='Test'):
        self._kwargs = {
            "bootstrap_servers": KafkaServerList,
            "client_id": ClientId,
            "acks": 1,
            "buffer_memory": 33554432,
            'compression_type': None,
            "retries": 3,
            "batch_size": 1048576,
            "linger_ms": 100,
            "key_serializer": lambda m: json.dumps(m, ensure_ascii=False).encode('utf-8'),
            "value_serializer": lambda m: json.dumps(m, ensure_ascii=False).encode('utf-8'),
        }
        self._topic = Topic
        try:
            self._producer = KafkaProducer(**self._kwargs)
        except Exception as err:
            print(err)

    def _onSendSucess(self, record_metadata):
        """
        :param record_metadata:
        :return:
        """
        print("发送成功")
        print("被发往的主题：", record_metadata.topic)
        # print("被发往的分区：", record_metadata.partition)
        # print("队列位置：", record_metadata.offset)

    def _onSendFailed(self):
        print("发送失败")

    def sendMessage(self, value=None, partition=None):
        if not value:
            return None
        kwargs = {
            "value": value,  # value 必须必须为字节或者被序列化为字节
            "key": None,  # 与value对应的键，可选
            "partition": partition  # 发送到哪个分区，整型
        }

        try:
            # future = self._producer.send(self._topic, **kwargs).add_callback(self._onSendSucess).add_errback(
            future = self._producer.send(self._topic, **kwargs)
            # self._onSendFailed)
            print("发送消息:", value)
        except KafkaTimeoutError as err:
            print(err)
        except Exception as err:
            print(err)

    def closeConnection(self, timeout=None):

        self._producer.close(timeout=timeout)

        def sendNow(self, timeout=None):
            try:
                self._producer.flush(timeout=timeout)
            except KafkaTimeoutError as err:
                print(err)
            except Exception as err:
                print(err)

def main():
    # p = Producer(KafkaServerList=["10.10.149.20:6667"], ClientId="Procucer01", Topic="original_news")
    p = Producer(KafkaServerList=["10.10.149.20:6667"], ClientId="Procucer01", Topic="js6class")
    for i in range(10):
        time.sleep(1)
        closePrice = random.randint(1, 500)
        msg = {
                "Publisher": "Procucer01",
                "股票代码": 60000 + i,
                "昨日收盘价": closePrice,
                "今日开盘价": 0,
                "今日收盘价": 0,
        }
        # p.sendMessage(value=msg)
        # """
        send_msg = {
                'key': None,
                'value': msg,
                'partition': None
        }
        try:
            p._producer.send('js6class', **send_msg)
        except Exception as e:
            print(e)
        # """
        # p.sendNow()
    p.closeConnection()

if __name__ == "__main__":
    try:
        main()
    finally:
        sys.exit()
